package com.pai4j.message.aggregation;

import com.pai4j.message.model.MessageContext;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 消息聚合器实现
 * 按规则聚合消息，减少消息数量
 *
 * @author PAI4J Team
 * @date 2025-11-01
 */
@Slf4j
@Component
@RequiredArgsConstructor
@ConditionalOnProperty(prefix = "pai4j.message.aggregation", name = "enabled", havingValue = "true", matchIfMissing = false)
public class MessageAggregatorImpl implements MessageAggregator {

    private final AggregationStrategy aggregationStrategy;

    /**
     * 待聚合消息池
     * key: 聚合键, value: 消息列表
     */
    private final Map<String, List<MessageContext>> pendingMessages = new ConcurrentHashMap<>();

    /**
     * 聚合规则列表
     */
    private final List<AggregationRule> rules = new ArrayList<>();

    /**
     * 消息添加时间
     * key: 聚合键, value: 首次添加时间
     */
    private final Map<String, LocalDateTime> addTimes = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        // 添加默认规则
        addRule(AggregationRule.createDefault());
        addRule(AggregationRule.createLikeRule());
        addRule(AggregationRule.createCommentRule());
        
        log.info("MessageAggregator 初始化完成，规则数量: {}", rules.size());
    }

    @Override
    public void add(MessageContext context) {
        if (context == null) {
            return;
        }

        try {
            // 查找匹配的规则
            AggregationRule matchedRule = findMatchingRule(context);
            if (matchedRule == null || !matchedRule.getEnabled()) {
                log.debug("消息不匹配任何聚合规则，跳过聚合");
                return;
            }

            // 生成聚合键
            String aggregationKey = matchedRule.generateAggregationKey(context);
            if (aggregationKey == null || aggregationKey.isEmpty()) {
                log.warn("生成聚合键失败，跳过聚合");
                return;
            }

            // 添加到待聚合池
            pendingMessages.computeIfAbsent(aggregationKey, k -> new ArrayList<>()).add(context);
            
            // 记录首次添加时间
            addTimes.putIfAbsent(aggregationKey, LocalDateTime.now());

            log.debug("消息添加到聚合池: key={}, count={}", 
                    aggregationKey, pendingMessages.get(aggregationKey).size());

            // 检查是否应该立即触发聚合
            List<MessageContext> pending = pendingMessages.get(aggregationKey);
            if (aggregationStrategy.shouldTrigger(pending.size(), matchedRule)) {
                if (pending.size() >= matchedRule.getMaxCount()) {
                    // 达到最大数量，立即触发
                    log.info("达到最大聚合数量，立即触发聚合: key={}, count={}", 
                            aggregationKey, pending.size());
                    trigger(aggregationKey);
                }
            }

        } catch (Exception e) {
            log.error("添加消息到聚合池失败: {}", e.getMessage(), e);
        }
    }

    @Override
    public List<MessageContext> getPendingMessages(String aggregationKey) {
        return pendingMessages.getOrDefault(aggregationKey, Collections.emptyList());
    }

    @Override
    public AggregatedMessage trigger(String aggregationKey) {
        if (aggregationKey == null || aggregationKey.isEmpty()) {
            return null;
        }

        List<MessageContext> messages = pendingMessages.remove(aggregationKey);
        addTimes.remove(aggregationKey);

        if (messages == null || messages.isEmpty()) {
            return null;
        }

        try {
            // 执行聚合
            AggregatedMessage aggregated = aggregationStrategy.aggregate(messages);
            
            if (aggregated != null) {
                log.info("消息聚合完成: key={}, count={}, content={}", 
                        aggregationKey, aggregated.getAggregateCount(), aggregated.getAggregatedContent());
            }

            return aggregated;

        } catch (Exception e) {
            log.error("消息聚合失败: key={}, error={}", aggregationKey, e.getMessage(), e);
            return null;
        }
    }

    @Override
    public List<AggregatedMessage> triggerAll() {
        List<AggregatedMessage> results = new ArrayList<>();

        // 复制键集合，避免并发修改
        Set<String> keys = new HashSet<>(pendingMessages.keySet());

        for (String key : keys) {
            AggregatedMessage aggregated = trigger(key);
            if (aggregated != null) {
                results.add(aggregated);
            }
        }

        log.info("批量触发聚合完成: count={}", results.size());
        return results;
    }

    @Override
    public List<AggregationRule> getRules() {
        return new ArrayList<>(rules);
    }

    @Override
    public void addRule(AggregationRule rule) {
        if (rule != null) {
            rules.add(rule);
            log.info("添加聚合规则: name={}, window={}s", rule.getRuleName(), rule.getWindowSeconds());
        }
    }

    @Override
    public void clear() {
        pendingMessages.clear();
        addTimes.clear();
        log.info("清除聚合池完成");
    }

    @Override
    public int getPendingCount() {
        return pendingMessages.values().stream()
                .mapToInt(List::size)
                .sum();
    }

    /**
     * 定时触发聚合（每分钟执行一次）
     */
    @Scheduled(cron = "0 * * * * ?")
    public void scheduledTrigger() {
        try {
            LocalDateTime now = LocalDateTime.now();
            List<String> keysToTrigger = new ArrayList<>();

            // 检查每个聚合键的时间窗口
            for (Map.Entry<String, LocalDateTime> entry : addTimes.entrySet()) {
                String key = entry.getKey();
                LocalDateTime addTime = entry.getValue();

                // 查找对应的规则
                AggregationRule rule = findRuleByKey(key);
                if (rule == null) {
                    continue;
                }

                // 计算时间差
                long seconds = java.time.Duration.between(addTime, now).getSeconds();
                
                // 超过时间窗口，触发聚合
                if (seconds >= rule.getWindowSeconds()) {
                    List<MessageContext> messages = pendingMessages.get(key);
                    if (messages != null && messages.size() >= rule.getMinCount()) {
                        keysToTrigger.add(key);
                    } else if (messages != null && messages.size() < rule.getMinCount()) {
                        // 未达到最小数量，清除
                        pendingMessages.remove(key);
                        addTimes.remove(key);
                        log.debug("聚合消息数量不足，清除: key={}, count={}", key, messages.size());
                    }
                }
            }

            // 触发聚合
            for (String key : keysToTrigger) {
                trigger(key);
            }

            if (!keysToTrigger.isEmpty()) {
                log.info("定时触发聚合完成: count={}", keysToTrigger.size());
            }

        } catch (Exception e) {
            log.error("定时触发聚合失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 查找匹配的规则
     */
    private AggregationRule findMatchingRule(MessageContext context) {
        return rules.stream()
                .filter(rule -> rule.matches(context))
                .findFirst()
                .orElse(null);
    }

    /**
     * 根据聚合键查找规则
     */
    private AggregationRule findRuleByKey(String key) {
        // 简单实现：使用默认规则
        // 实际可以根据key的格式判断使用哪个规则
        return rules.isEmpty() ? null : rules.get(0);
    }
}

